package com.qq.rocketmq.consumer;

import com.qq.rocketmq.constant.ProducerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

@Slf4j
public abstract class AbstractRocketMqComsumer implements InitializingBean {
    @Autowired
    private  ProducerConfig  producerConfig;

    @Override
    public void afterPropertiesSet() throws Exception {
        if (StringUtils.isBlank(getTopic()) ){
            throw  new RuntimeException("请配置topic,tag,consumerGroup .........");
        }

        log.info("Rocketmq consumer starting, topic:[{}], namesrvAddr[{}],groupName:[{}]",getTopic(),producerConfig.getNamesrvAddr(),producerConfig.getGroupName());

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(producerConfig.getGroupName());

        consumer.setNamesrvAddr(producerConfig.getNamesrvAddr());

        consumer.setConsumerGroup(producerConfig.getGroupName());

        consumer.subscribe(this.getTopic(), getMessageSelector());

        consumer.setConsumeMessageBatchMaxSize(getBatchMaxSize());

        // 如果非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setMessageModel(getMessageMode());
        // 开启内部类实现监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                return dealBody(msgs);
            }
        });
        consumer.start();
        log.info("rocketmq consumer启动成功---------------------------------------");

    }

    /**
     * 处理的业务逻辑
     * @param msgs
     * @return
     */
    protected abstract ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs);

    /**
     * 默认按tag 的方式过滤，若要按sql的方式过滤重写此方法
     * @return
     */
    protected MessageSelector getMessageSelector(){
        return MessageSelector.byTag(getTag());
    }

    /**
     * 获取mq的topic
     * @return
     */
    protected abstract  String getTopic();

    /**
     * 需要过滤的tag，默认为所有
     * @return String
     */
    protected  String getTag(){
        return "*";
    };

    /**
     * 获取每次处理的条数
     * @return Integer
     */
    protected  Integer getBatchMaxSize(){
        return 1;
    };

    /**
     * RocketMq消息消费的模式 默认为集群消费 ,若要设置为广播模式 重新实现该方法
     * @return MessageModel
     */
    protected MessageModel getMessageMode(){
        return MessageModel.CLUSTERING;
    }

}
